package cc.idoone.spring_boot_hello.amqp.sendconfirm;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author hemj
 * @since 2017/10/15
 */
@Configuration
public class AmqpConfig {

	static final String CB_ROUTINGKEY = "spring-boot-cbkey";
	static final String CB_QUENE = "spring-boot-cbqueue";
	static final String CB_EXCHANGE = "spring-boot-cbexchange";

	@Bean
	Queue cbQueue() {
		return new Queue(CB_QUENE, true);
	}

	@Bean
	TopicExchange cbExchange() {
		return new TopicExchange(CB_EXCHANGE);
	}

	@Bean
	Binding cbBinding(@Qualifier("cbQueue") Queue queue, @Qualifier("cbExchange") TopicExchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with(CB_ROUTINGKEY);
	}

	@Bean
	SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory, @Qualifier("cbQueue") Queue queue) {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
		//container.setQueues(queue);
		container.setQueueNames(CB_QUENE);
		container.setExposeListenerChannel(true);
		container.setMaxConcurrentConsumers(1);
		container.setConcurrentConsumers(1);
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
		container.setMessageListener(new ChannelAwareMessageListener() {

			// 接收方式3
			@Override
			public void onMessage(Message message, Channel channel) throws Exception {
				byte[] body = message.getBody();
				System.out.println("收到消息: " + new String(body));
				channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费
				//throw new Exception("接收异常");
			}
		});
		return container;
	}
}
